package com.linkstec.kafka;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.Arrays;

/**
 * 消息路由
 *
 * @author linkage
 *
 */
public class KafkaRouterPriority extends KafkaConsumerServicePriority {
	private static final Logger logger = LogManager.getLogger();

   static String PRIORITY_MESSAGE_TOPIC = null;
   static String PRIORITY_ROUTER_GROUPID = null;

   static {

        String routerTopic = "TEST";//ConfigUtil.getProperty("priority.router.topic");
        if (StringUtils.isEmpty(routerTopic)) {
            throw new RuntimeException("priority.router.topic未配置，推送服务退出运行");
        }
        PRIORITY_MESSAGE_TOPIC = routerTopic;
      String routerGroupId = "testGroup";//ConfigUtil.getProperty("priority.router.groupId");
      if (StringUtils.isEmpty(routerGroupId)) {
         throw new RuntimeException("priority.router.groupId未配置，推送服务退出运行");
      }
      PRIORITY_ROUTER_GROUPID = routerGroupId;
    }
   @Override
   public void run() {
      for (int j = 1; j <= PRIORITY_EVENT_FLOW_NUM; j++) {
         String topic = PRIORITY_MESSAGE_TOPIC + PRIORITY_TOPIC_DELIMITER + (j);
         logger.info("订阅topic:" + topic);
         consumers.get(j).subscribe(Arrays.asList(topic));
      }
      while (true) {
         logger.info("调度一次");
         /*List<String> listResult = */poll();
         /*for (String result : listResult) {
        	 logger.info(result);
         }*/
      }
   }
}